package mapreduce;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Scanner;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import util.hadoop.HDFSUtil;
import util.init.Location;
import util.ip.IpReg;
import util.ip.ProFilter;
import util.math.IntervalCalc;
import util.math.MyMath;

/*包的格式
 * "No.","Time","Source","Destination","Protocol","Source Port","Destination Port","Length"
	"1","0.000000","169.254.34.247","224.0.0.252","LLMNR","54663","5355","75"
	"2","0.102452","10.10.102.30","224.0.0.252","LLMNR","56746","5355","66"
	"3","0.103599","169.254.34.247","224.0.0.252","LLMNR","54663","5355","75"
 */


/*
 * datainput/type3in目录下的文件为输入，输出文件在dataoutput/type4out目录下，
 * 为每个流的一些特征值
 */
public class ProcessType4 {
	
	public static Set<String> userSet;
	
	public static String locationRegex;
	
	public static class MapClass extends Mapper<LongWritable, Text, Text, Text> {
		
		IpReg ipReg = new IpReg();
		ProFilter proFilter = new ProFilter();
		
		@Override
		protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			/*
			 * 过滤南京图书馆用户地址ip的正则表达式
			 */
		
			ipReg.setSrcPattern(locationRegex);
			context.nextKeyValue();
		}

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
				throws IOException, InterruptedException {
			

			// TODO Auto-generated method stub
			
				String[] line = value.toString().split(",");
				
				String source = line[2].replaceAll("\"", "");
				String destination = line[3].replaceAll("\"", "");
				//String protocol = line[4].replaceAll("\"", "");
				
				//System.out.println(ipReg.isUserSrc(source) && ipReg.isISP(destination) && proFilter.isLegal(protocol) );
				//if (ipReg.isUserSrc(source) && ipReg.isISP(destination) && proFilter.isLegal(protocol) )  {
				if (ipReg.isUserSrc(source) && ipReg.isISP(destination))  {
						String time = line[1].replaceAll("\"", "");
						String length = line[7].replaceAll("\"", "");	
						Text src = new Text(source);	
						String mapOut = time + "," + length;
						context.write(src, new Text(mapOut));
				}
		}
	}
	/*
	 * 对src处理，没有注释
	 */
	public static class ReduceClass extends Reducer<Text, Text, NullWritable, Text> {
		MultipleOutputs<NullWritable, Text> mos;
		int ipNo = 0;
		MyMath mm;
		MyMath mm1;
		List<Double> lenList = new ArrayList<Double>();
		List<Double> timeList = new ArrayList<Double>();
		@Override
		protected void setup(Reducer<Text, Text, NullWritable, Text>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			String out = "Ip no.,Src Ip,Number of packets,Min,Max,Average,Median,Standard,Skewness,Kurtosis\r";
			mos = new MultipleOutputs<>(context);
			mos.write("type4", NullWritable.get(), new Text(out), "PacketInterArrivalTime");
			mos.write("type4", NullWritable.get(), new Text(out), "PacketSize");
		}

		@Override
		protected void reduce(Text arg0, Iterable<Text> arg1,
				Reducer<Text, Text, NullWritable, Text>.Context arg2)
						throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			
			
			for (Text text : arg1) {	
				String[] temp = text.toString().split(",");
				double time = Double.valueOf(temp[0]);
				double length = Double.valueOf(temp[1]);
				lenList.add(length);
				timeList.add(time);
			}
			if (userSet.contains(arg0.toString())) {
				ipNo++;
				mm = new MyMath(lenList);
				double minS = mm.getMin();
				double maxS = mm.getMax();
				double aveS = mm.getAve();
				double medianS = mm.getMedian();
				double standardS = mm.getStandard();
				double skewS = mm.getSkew();
				double kurtosisS = mm.getKurt();
				int packetNumS = mm.getSize();
				
				
				
				IntervalCalc intervalCalc = new IntervalCalc(timeList);
				intervalCalc.setIntervalTimeList();
				mm1 = new MyMath(intervalCalc.getIntervalTimeList());
				if (mm1.getSize() == 0) {
					System.out.println(arg0.toString());
				}
				double minT = mm1.getMin();
				double maxT = mm1.getMax();
				double aveT = mm1.getAve();
				double medianT = mm1.getMedian();
				double standardT = mm1.getStandard();
				double skewT = mm1.getSkew();
				double kurtosisT = mm1.getKurt();
				int packetNumT = mm1.getSize() + 1;
				
				String outS  = ipNo + "," + arg0.toString() + "," 
						+ packetNumS + "," + minS + "," + maxS
						+ "," + aveS + "," + medianS + "," 
						+ standardS + "," + skewS + "," + kurtosisS + "\r";
				mos.write("type4", NullWritable.get(), new Text(outS), "PacketSize");
				
				String outT  = ipNo + "," + arg0.toString() + "," 
							+ packetNumT + "," + minT + "," + maxT
							+ "," + aveT + "," + medianT + "," 
							+ standardT + "," + skewT + "," + kurtosisT + "\r";
				mos.write("type4", NullWritable.get(), new Text(outT), "PacketInterArrivalTime");
				
			}
			lenList.clear();
			timeList.clear();
		}

		@Override
		protected void cleanup(Reducer<Text, Text, NullWritable, Text>.Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			mos.close();
		}
		
		
		
		
	}
	
	
	
	

	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Location init = new Location();
		System.out.println("请输入需要过滤的地址：");
		Scanner scanner = new Scanner(System.in);  
        String location = scanner.nextLine();
		locationRegex = init.getLocationRegex(location);
		
		String master = args[0];
		Configuration conf = new Configuration();
		conf.set("fs.defaultFS", "hdfs://" + master + ":9000");
		
		//conf.set("mapreduce.input.keyvaluelinerecordreader.key.value.separator", ",");

		HDFSUtil hdfsUtil = new HDFSUtil(conf);
		Job job = Job.getInstance(conf, "process");
		
		job.setJarByClass(ProcessType4.class);
		
		job.setMapperClass(MapClass.class);
		job.setReducerClass(ReduceClass.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		
		MultipleOutputs.addNamedOutput(job, "type4", TextOutputFormat.class, NullWritable.class, Text.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		//String location = "njlib";
		Path in  = new Path("hdfs://" + master +":9000/datainput/" + location +"/");
		Path out = new Path("hdfs://" + master + ":9000/dataoutput/type4/");
		hdfsUtil.delete(out);
		userSet = hdfsUtil.getUserIp();
		
		FileInputFormat.addInputPath(job, in);
		FileOutputFormat.setOutputPath(job, out);
		System.out.println(job.waitForCompletion(true) ? 0 : 1);
		Path out1 = new Path("hdfs://" + master + ":9000/dataoutput/type4/_SUCCESS");
		Path out2 = new Path("hdfs://" + master + ":9000/dataoutput/type4/part-r-00000");
		hdfsUtil.delete(out1);
		hdfsUtil.delete(out2);

	}

}
